package com.hanxiaozhang.example.listener.mgsconsumer;

/**
 * 〈一句话功能简述〉<br>
 * 〈〉
 *
 * @author hanxinghua
 * @create 2022/9/29
 * @since 1.0.0
 */

import com.hanxiaozhang.constant.RocketConstant;
import com.hanxiaozhang.example.entity.RocketMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Service;


/**
 * 功能描述: <br>
 * 〈Push模式消费，配置消费端消费起始位点〉
 *
 * @Author: hanxinghua
 * @Date: 2022/10/5
 */
@Slf4j
@Service
@RocketMQMessageListener(topic = RocketConstant.CONSUME_POINT_TOPIC, consumerGroup = RocketConstant.CONSUME_POINT_CONSUMER_GROUP)
public class No9ConsumePointListener implements RocketMQListener<RocketMessage>, RocketMQPushConsumerLifecycleListener {

    /**
     * 客户端收到的消息
     * @param message
     */
    @Override
    public void onMessage(RocketMessage message) {
        log.info("{}收到消息：{}", this.getClass().getSimpleName(), message);
    }

    /**
     * 对消费者客户端的一些配置
     * @param consumer
     */
    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {

        // 设置消费点的类型：消费者客户端将从指定的时间戳开始
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
        // 启动之后10s
        consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis())+10*1000);

    }
}
